package com.gukun.springboot.tdengine.service.writeplant;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;


public class FastWriteTagData {
    static int taskQueueCapacity = 1000000;
    static Map<String,List<BlockingQueue<String>>> taskQueuesMap = new HashMap<>();
    static Map<String,List<ReadTask>> readTasksMap = new HashMap<>();
    static Map<String,List<WriteTask>> writeTasksMap = new HashMap<>();
    DataBaseMonitor databaseMonitor = new DataBaseMonitor();

    private Boolean isCreateDatabase = false;

    public void stopAll() {
        if(readTasksMap != null){
            Iterator<String> its = readTasksMap.keySet().iterator();
            while(its.hasNext()){
                String key = its.next();
                List<ReadTask> readTasks = readTasksMap.get(key);
                if(readTasks != null){
                    readTasks.forEach(task -> task.stop());
                }
            }
        }
        if(writeTasksMap != null){
            Iterator<String> its = writeTasksMap.keySet().iterator();
            while(its.hasNext()){
                String key = its.next();
                List<WriteTask> writeTasks = writeTasksMap.get(key);
                if(writeTasks != null){
                    writeTasks.forEach(task -> task.stop());
                }
            }
        }
        databaseMonitor.close();

        readTasksMap = new HashMap<>();
        writeTasksMap = new HashMap<>();
        taskQueuesMap = new HashMap<>();
    }

    public void stopAll(String keyNameId) {
        List<ReadTask> readTasks = readTasksMap.get(keyNameId);
        if(readTasks != null){
            readTasks.forEach(task -> task.stop());
        }
        readTasksMap.remove(keyNameId);

        List<WriteTask> writeTasks = writeTasksMap.get(keyNameId);
        if(writeTasks != null){
            writeTasks.forEach(task -> task.stop());
        }
        writeTasksMap.remove(keyNameId);
        if(readTasksMap.keySet().size() == 0 && writeTasksMap.keySet().size()==0){
            databaseMonitor.close();
        }
        if(taskQueuesMap.containsKey(keyNameId)){
            taskQueuesMap.remove(keyNameId);
        }
    }

    public FastWriteTagData(TDConfigDto tdConfigDto,Boolean isCreateDatabase,String keyNameId) throws Exception {
            this.isCreateDatabase = isCreateDatabase;
            init(tdConfigDto,keyNameId);
    }

    /**
     *
     * @param tdConfigDto 连接配置对象
     * @throws InterruptedException
     * @throws SQLException
     */
    public void init(TDConfigDto tdConfigDto,String keyNameId) throws Exception {
        if(tdConfigDto.getTablePre() != null && tdConfigDto.getTablePre() != ""){
            int readTaskCount = 1;
            int writeTaskCount = 3;
            int maxBatchSize = 100000;
            if(tdConfigDto.getReadTaskCount() != null){readTaskCount = tdConfigDto.getReadTaskCount();}
            if(tdConfigDto.getWriteTaskCount() != null){writeTaskCount = tdConfigDto.getWriteTaskCount();}
            if(tdConfigDto.getMaxBatchSize() != null){maxBatchSize = tdConfigDto.getMaxBatchSize();}
            String jdbcURL = "jdbc:TAOS-WS://"+tdConfigDto.getIp()+":"+tdConfigDto.getPort()+"?charset=UTF-8&timezone=UTC-8&" +
                    "user="+tdConfigDto.getUsername()+"&password="+tdConfigDto.getPassword();
            if(tdConfigDto.getPort() == 6030){
                jdbcURL = "jdbc:TAOS://"+tdConfigDto.getIp()+":"+tdConfigDto.getPort()+"?charset=UTF-8&timezone=UTC-8&" +
                        "user="+tdConfigDto.getUsername()+"&password="+tdConfigDto.getPassword();
            }
//            if(this.isCreateDatabase){
//                //数据库已有别的项目创建。所以在这里等待片刻
//                try{
//                    Thread.sleep(5000);
//                }catch(Exception e){
//
//                }
//            }else{
//                //数据库未创建，所以需要进行创建
//                databaseMonitor.init(jdbcURL).prepareDatabase(tdConfigDto,isCreateDatabase);
//            }
            databaseMonitor.init(jdbcURL).prepareDatabase(tdConfigDto,isCreateDatabase);
            System.out.println("创建线程"+writeTaskCount);
            List<WriteTask> writeTasksList = new ArrayList<>();
            List<BlockingQueue<String>> taskQueuesList = new ArrayList<>();
            for (int i = 0; i < writeTaskCount; ++i) {
                BlockingQueue<String> queue = new ArrayBlockingQueue<>(taskQueueCapacity);
                taskQueuesList.add(queue);
                WriteTask task = new WriteTask(queue, maxBatchSize,jdbcURL,tdConfigDto.getDbName());
                writeTasksList.add(task);
                Thread t = new Thread(task);
                t.setName("WriteThread-" + i);
                t.start();
            }
            taskQueuesMap.put(keyNameId,taskQueuesList);
            writeTasksMap.put(keyNameId,writeTasksList);
//            for (int i = 0; i < readTaskCount; ++i) {
//                ReadTask task = new ReadTask(i, taskQueues,tdConfigDto);
//                readTasks.add(task);
//                Thread t = new Thread(task);
//                t.setName("ReadThread-" + i);
//                t.start();
//            }
        }
    }

    /**
     * 获取待添加数据的队列
     * @return
     */
    public List<BlockingQueue<String>> getTaskQueues(String keyNameId) {
        if (taskQueuesMap.containsKey(keyNameId)){
            return taskQueuesMap.get(keyNameId);
        }
        return new ArrayList<>();
    }
}
